package com.biz.aoao.mysql.mysqlbinlogsync.listener;

import com.biz.aoao.mysql.mysqlbinlogsync.MySqlBinLogSyncConstants;
import com.biz.aoao.mysql.mysqlbinlogsync.extra.IBinlogSyncInfoSetting;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * @Author scgaopan
 * @Date 2020/8/5 20:06
 * @Version 1.0
 * @Desc  自定义mysql binlog监听器
 */
@Slf4j
@Component
public class CustomerEventListener implements BinaryLogClient.EventListener{

    @Autowired
    private IBinlogSyncInfoSetting iBinlogSyncInfoSetting;
    @Override
    public void onEvent(Event event) {
        EventHeaderV4 header = event.getHeader();
        if(header==null){
            return;
        }
        EventType eventType = header.getEventType();
        String key=null;
        String fileName=null;
        Long nextPosition=header.getNextPosition();
        Long eventLength=header.getEventLength();
        Long serverId=header.getServerId();

        switch (eventType){
            case ROTATE:
                RotateEventData rotateEventData=event.getData();
                fileName=rotateEventData.getBinlogFilename();
                nextPosition=rotateEventData.getBinlogPosition();
                break;
            case TABLE_MAP:
                TableMapEventData data = event.getData();
                key = data.getDatabase().concat(MySqlBinLogSyncConstants.DATEBASE_TABLE_FLEX).concat(data.getTable());
               if(MySqlBinLogSyncConstants.TABLE_INFO_MAPPING.containsKey(key)){
                   MySqlBinLogSyncConstants.TABLE_NAME_ID_MAPPING.put(data.getTableId(),key);
               }
                break;
            case EXT_UPDATE_ROWS:
                UpdateRowsEventData updateRowsEventData= event.getData();
                key=MySqlBinLogSyncConstants.TABLE_NAME_ID_MAPPING.get(updateRowsEventData.getTableId());
                if(key==null){
                   return;
                }
                MySqlBinLogSyncConstants.TABLE_INFO_MAPPING.get(key).invokeUpdate(updateRowsEventData.getRows());
                break;
            case EXT_WRITE_ROWS:
                WriteRowsEventData writeRowsEventData=event.getData();
                key=MySqlBinLogSyncConstants.TABLE_NAME_ID_MAPPING.get(writeRowsEventData.getTableId());
                if(key==null){
                    return;
                }
                MySqlBinLogSyncConstants.TABLE_INFO_MAPPING.get(key).invokeInsert(writeRowsEventData.getRows());
                break;
            case EXT_DELETE_ROWS:
                DeleteRowsEventData deleteRowsEventData=event.getData();
                key=MySqlBinLogSyncConstants.TABLE_NAME_ID_MAPPING.get(deleteRowsEventData.getTableId());
                if(key==null){
                    return;
                }
                MySqlBinLogSyncConstants.TABLE_INFO_MAPPING.get(key).invokeDelete(deleteRowsEventData.getRows());
                break;
                default:

                break;
        }
        //mysql日志位置的记录
        iBinlogSyncInfoSetting.recordingBinLogPositon(serverId,fileName,eventLength,nextPosition,eventType);
    }
}
